package site.teamo.biu.flink.sdk.job;

import cn.hutool.core.annotation.AnnotationUtil;
import lombok.Getter;
import site.teamo.biu.flink.sdk.annotation.function.BFProcessFunction;
import site.teamo.biu.flink.sdk.annotation.function.BFSinkFunction;
import site.teamo.biu.flink.sdk.annotation.function.BFSourceFunction;

/**
 * @author haocongshun
 * @date 2023/07/26 11:04:52
 */
@Getter
public class JobNode {
    private JobNodeInfo info;

    private final Class<?> functionClass;

    private Class<?> inClass;

    private Class<?> outClass;

    public JobNode(JobNodeInfo info) {
        try {
            this.info = info;
            functionClass = Class.forName(this.info.getFunction());
            if (AnnotationUtil.hasAnnotation(functionClass, BFProcessFunction.class)) {
                BFProcessFunction annotation = functionClass.getAnnotation(BFProcessFunction.class);
                inClass = annotation.in();
                outClass = annotation.out();
            } else if (AnnotationUtil.hasAnnotation(functionClass, BFSourceFunction.class)) {
                BFSourceFunction annotation = functionClass.getAnnotation(BFSourceFunction.class);
                outClass = annotation.out();
            } else if (AnnotationUtil.hasAnnotation(functionClass, BFSinkFunction.class)) {
                BFSinkFunction annotation = functionClass.getAnnotation(BFSinkFunction.class);
                outClass = annotation.in();
            } else {
                throw new RuntimeException("未发现function注解");
            }
        } catch (Exception e) {
            throw new RuntimeException("Job Node 构造失败", e);
        }
    }

    public String toCode(String parentName) {
        switch (functionClass.getSuperclass().getName()) {
            case "site.teamo.biu.flink.sdk.function.AbstractSourceFunction":
                return toSourceFunctionCode();
            case "site.teamo.biu.flink.sdk.function.AbstractProcessFunction":
                return toProcessFunctionCode(parentName);
            case "site.teamo.biu.flink.sdk.function.AbstractSinkFunction":
                return toSinkFunctionCode(parentName);
            default:
                throw new RuntimeException("不支持的function实现: " + functionClass.getSuperclass().getName());
        }
    }

    private String toSourceFunctionCode() {
        return String.format("org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator<%s> %s =environment" +
                        ".addSource(new %s(\"%s\", context))" +
                        ".name(\"%s\")" +
                        ".setParallelism(%s)" +
                        ".returns(%s.class);",
                outClass.getName(),
                info.getName(),
                functionClass.getName(),
                info.getName(),
                info.getName(),
                info.getParallelism(),
                outClass.getName());
    }

    private String toProcessFunctionCode(String parentName) {
        return String.format("org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator<%s> %s = %s" +
                        ".process(new %s(\"%s\", context))" +
                        ".name(\"%s\")" +
                        ".setParallelism(%s)" +
                        ".returns(%s.class);",
                outClass.getName(),
                info.getName(),
                parentName,
                functionClass.getName(),
                info.getName(),
                info.getName(),
                info.getParallelism(),
                outClass.getName());
    }

    private String toSinkFunctionCode(String parentName) {
        return String.format("%s.addSink(new %s(\"%s\", context)).name(\"%s\").setParallelism(%s);",
                parentName,
                functionClass.getName(),
                info.getName(),
                info.getName(),
                info.getParallelism());
    }
}
